package org.huel.dataprocessing

import org.apache.kafka.clients.producer.{KafkaProducer, ProducerConfig, ProducerRecord}
import org.apache.kafka.common.serialization.StringDeserializer
import org.apache.spark.SparkConf
import org.apache.spark.streaming.kafka010.ConsumerStrategies.Subscribe
import org.apache.spark.streaming.kafka010.KafkaUtils
import org.apache.spark.streaming.kafka010.LocationStrategies.PreferConsistent
import org.apache.spark.streaming.{Seconds, StreamingContext}
import org.json.JSONObject

import java.util

/**
 * @Description TODO
 * @author Liweijian.
 * @date 2024/11/26 17:51
 */
object runSSC {
  def main(args: Array[String]): Unit = {

    /*
    * 消费者配置项
    * */
    // 创建 SparkConf
    val conf = new SparkConf().setAppName("runssc").setMaster("local[*]")

    val ssc = new StreamingContext(conf, Seconds(2)) // 微批次处理，间隔时间
    ssc.sparkContext.setLogLevel("error") // 设置日志级别

    // 配置 Kafka 参数
    val kafkaParams = Map[String, Object](
      "bootstrap.servers" -> "43.143.125.94:9092",
      "key.deserializer" -> classOf[StringDeserializer],
      "value.deserializer" -> classOf[StringDeserializer],
      "group.id" -> "attendance1" // 将 groupId 作为参数传入
    )

    val topicName = Array("attendance")
    val streamRdd = KafkaUtils.createDirectStream[String,String](
      ssc,
      PreferConsistent,
      Subscribe[String,String](topicName, kafkaParams)
    )

    /*
    * 生产者配置项（推送到新主题
    * */
    val producerProperties = new util.HashMap[String, Object]()
    producerProperties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "43.143.125.94:9092")
    producerProperties.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")
    producerProperties.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer")

    /*
    * CourseCount
    * */
    val result = streamRdd.map(_.value())
      .map(line => {
        val parts = line.split("\t")
        val course = parts(2) // 获取课程名称
        (course, 1) // 返回元组 ((课程, 状态), 1)
      })
      .reduceByKeyAndWindow(_ + _, Seconds(2), Seconds(2)) // 按课程和状态统计每2秒的数据

    //println("======")
    // 将结果输出到控制台并发送到Kafka
    result.foreachRDD(rdd => {
      rdd.foreach {
        case (course, count) => {
          // 创建 JSON 对象
          val json = new JSONObject()
          json.put("course", course)
          json.put("count", count)

          val jsonString = json.toString() // 转换为字符串
          //println("========")
          println(jsonString) // 打印 JSON 字符串

          // 连接新的客户端
          val producer = new KafkaProducer[String, String](producerProperties)
          // 把 JSON 数据发送到指定的 Kafka 主题
          producer.send(new ProducerRecord[String, String]("CourseCount1", jsonString))
          producer.close() // 关闭生产者
        }
      }
    })

    /*
    * LA
    * */
    val result1 = streamRdd.map(_.value())
      .map(line => {
        val parts = line.split("\t")
        val status = parts(4) // L 或 A
        (status, 1) // 返回元组 (状态, 1)
      })
      .reduceByKeyAndWindow(_ + _, Seconds(2), Seconds(2)) // 设置窗口时间，统计每2秒的数据

    // 将结果输出到控制台并发送到 Kafka
    result1.foreachRDD {
      rdd =>
        rdd.foreach { case (status, count) =>
          // 创建 JSON 对象
          val json = new JSONObject()
          json.put("status", status)
          json.put("count", count)

          val jsonString = json.toString() // 转换为字符串
          println(jsonString)

          // 连接新的客户端
          val producer = new KafkaProducer[String, String](producerProperties)
          // 把 JSON 数据发送到指定的 Kafka 主题
          producer.send(new ProducerRecord[String, String]("LACount", jsonString))
          producer.close() // 关闭生产者
        }
    }


    /*
    * LAByClass
    * */
    val result2 = streamRdd.map(_.value())
      .map(line => {
        val parts = line.split("\t") // 根据空白字符分割
        val classId = parts(0) // 班级号
        val status = parts(4) // L 或 A
        ((classId, status), 1) // 返回元组 ((班级号, 状态), 1)
      })
      .reduceByKeyAndWindow(_ + _, Seconds(2), Seconds(2)) // 设置窗口时间，统计每2秒的数据

    // 将结果格式化并输出到控制台并发送到Kafka
    result2.foreachRDD {
      rdd => {
        //println("="*20)
        rdd.foreach {
          case ((classId, status), count) => {
            // 创建 JSON 对象
            val json = new JSONObject()
            json.put("classId", classId)
            json.put("status", status)
            json.put("count", count)

            val jsonString = json.toString() // 转换为字符串
            println(jsonString) // 打印 JSON 字符串

            // 连接新的客户端
            val producer = new KafkaProducer[String, String](producerProperties)
            // 把 JSON 数据发送到指定的 Kafka 主题
            producer.send(new ProducerRecord[String, String]("LACountByClass1", jsonString))
            producer.close() // 关闭生产者
          }
        }
      }
    }

    //开始scc
    ssc.start()
    ssc.awaitTermination()//监控
  }
}
